package com.dyna.schedule;

import com.alibaba.fastjson.JSON;
import com.dyna.constants.RedisKeyConstant;
import com.dyna.core.exception.CustomException;
import com.dyna.entity.DelayMessage;
import com.dyna.factory.RedisDelayMsgHandlerFactory;
import com.dyna.handler.RedisDelayMsgHandler;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Optional;
import java.util.Set;

/**
 * 监听ZSet中的延时消息
 */
@Service
public class DelayQueueListenerTask {

    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private RedisDelayMsgHandlerFactory messageHandlerFactory;

    /**
     * 处理到期消息（定时任务）
     */
    @Scheduled(fixedRate = 1000)
    public void processExpiredMessages() {
        long now = System.currentTimeMillis();

        // 原子操作获取并删除消息
        Set<String> messages = stringRedisTemplate.opsForZSet().rangeByScore(
                RedisKeyConstant.DELAY_QUEUE_KEY, 0, now);

        assert messages != null;
        if (!messages.isEmpty()) {
            // 处理消息
            messages.forEach(msg -> {
                DelayMessage delayMessage = JSON.parseObject(msg, DelayMessage.class);
                handleMessage(delayMessage);
            });

            // 删除已处理消息
            stringRedisTemplate.opsForZSet().removeRangeByScore(
                    RedisKeyConstant.DELAY_QUEUE_KEY, 0, now);
        }
    }

    private void handleMessage(DelayMessage message) {
        Integer messageType = Optional.ofNullable(message).map(DelayMessage::getType).orElseThrow(() -> new CustomException("未获取到消息类型"));
        // 假设DelayMessage中有getType方法
        RedisDelayMsgHandler handler = messageHandlerFactory.getHandler(messageType);
        if (handler != null) {
            handler.handle(message);
        } else {
            throw new CustomException("未找到对应的消息处理器");
        }
    }

}